package com.example.myapplication;

import androidx.annotation.NonNull;

import java.util.ArrayList;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;

public class CreateOperatorDemo {
        Observer<Object> onComplete = new Observer<Object>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("onSubscribe....");
            }

            @Override
            public void onNext(@NonNull Object o) {
                System.out.println("onNext...." + o.toString());
            }

            @Override
            public void onError(@NonNull Throwable e) {
                System.out.println("onError" + e.toString());
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        };



    //    创建观察者的用法
    public void test1(){
        Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Object> emitter) throws Exception {
//事件产生的地方
                emitter.onNext("1");
                emitter.onNext("222");
                emitter.onNext("aaa");
                emitter.onError(new Throwable("Errrer"));
//                emitter.onComplete();
                /**
                 * 这个方法用来做网络操作或者说是耗时的操作
                 * 异步操作
                 */
            }
        }).subscribe(onComplete);
    }
//    创建消费者的用法
public void test2(){
    Observable.create(new ObservableOnSubscribe<Object>() {
        @Override
        public void subscribe(@NonNull ObservableEmitter<Object> emitter) throws Exception {
//事件产生的地方
            emitter.onNext("1");
            emitter.onNext("222");
            emitter.onNext("aaa");
            emitter.onError(new Throwable("Errrer"));
            emitter.onComplete();
        }
    }).subscribe(new Consumer<Object>() {
        @Override
        public void accept(Object o) throws Exception {
            System.out.println("accept"+o.toString());
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            System.out.println("accept"+throwable.toString());
        }
    });
}
public void testJust(){
        Observable.just("1","22","33","44","aa")
                .subscribe(onComplete);
}
//无参数个数要求
public void testfromArray(){
        Observable.fromArray("1","2","3","4","5","6","7","8","9","10")
                .subscribe(onComplete);
}
public void test4(){
    ArrayList<String> list =new ArrayList<>();
    list.add("22");
    list.add("22d");
        Observable.fromIterable(list)
                .subscribe(onComplete);
}


}
